package com.hui.config;

import com.hui.listener.JobCompletionNotificationListener;
import com.hui.domain.Person;
import com.hui.processor.PersonItemProcessor;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.tomcat.jdbc.pool.DataSource;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.batch.MyBatisBatchItemWriter;
import org.mybatis.spring.batch.MyBatisPagingItemReader;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.file.transform.FixedLengthTokenizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.core.BeanPropertyRowMapper;

import java.util.HashMap;

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    // tag::readerwriterprocessor[]
    @Bean
    public FlatFileItemReader<Person> flatFileItemReader() {
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
        reader.setResource(new ClassPathResource("sample-data.csv"));
        FixedLengthTokenizer fixedLengthTokenizer = new FixedLengthTokenizer();
        reader.setLineMapper(new DefaultLineMapper<Person>() {{
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setNames(new String[]{"firstName", "lastName"});
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                setTargetType(Person.class);
            }});
        }});
        return reader;
    }

    @Bean
    public JdbcPagingItemReader<Person> jdbcPagingItemReader(DataSource dataSource) {
        JdbcPagingItemReader<Person> reader = new JdbcPagingItemReader<>();
        reader.setDataSource(dataSource);
        reader.setFetchSize(100);

        reader.setQueryProvider(new MySqlPagingQueryProvider() {{
            setSelectClause("SELECT person_id,first_name,last_name");
            setFromClause("from people");
            setWhereClause("last_name=:lastName");
            setSortKeys(new HashMap<String, Order>() {{
                put("person_id", Order.ASCENDING);
            }});
        }});
        reader.setParameterValues(new HashMap<String, Object>() {{
            put("lastName", "DOE");
        }});
        reader.setRowMapper(new BeanPropertyRowMapper<>(Person.class));
        return reader;
    }

    @Bean
    public MyBatisPagingItemReader<Person> myBatisPagingItemReader(SqlSessionFactory sqlSessionFactory) {
        MyBatisPagingItemReader<Person> reader = new MyBatisPagingItemReader<>();
        reader.setSqlSessionFactory(sqlSessionFactory);
        return reader;
    }


    @Bean
    public JdbcBatchItemWriter<Person> jdbcBatchItemWriter(DataSource dataSource) {
        return new JdbcBatchItemWriter<Person>() {{
            setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
            setSql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)");
            setDataSource(dataSource);
        }};
    }

/*    @Bean
    public FlatFileItemWriter<Person> flatFileItemWriter(DataSource dataSource) {
        //TODO yhq
        FlatFileItemWriter<Person> writer = new FlatFileItemWriter<>();
        writer.setAppendAllowed(true);
        writer.setEncoding("UTF-8");
//        writer.set(dataSource);

        return writer;
    }*/

    @Bean
    public MyBatisBatchItemWriter<Person> myBatisBatchItemWriter(SqlSessionFactory sqlSessionFactory) {
        MyBatisBatchItemWriter<Person> itemWriter = new MyBatisBatchItemWriter<>();
        itemWriter.setSqlSessionFactory(sqlSessionFactory);
        return itemWriter;
    }


    // end::readerwriterprocessor[]

    // tag::jobstep[]
    @Bean
    public Job importUserJob(JobBuilderFactory jobBuilderFactory, JobCompletionNotificationListener listener, Step step) {
        return jobBuilderFactory.get("importUserJob")
                .incrementer(new RunIdIncrementer())
                .listener(listener)
                .start(step)
                .build();
    }

    @Bean
    public Step step1(StepBuilderFactory stepBuilderFactory, PersonItemProcessor processor, ItemWriter jdbcBatchItemWriter, ItemReader flatFileItemReader) {
        /*CompositeItemProcessor compositeItemProcessor = new CompositeItemProcessor();
        compositeItemProcessor.setDelegates(Lists.newArrayList(processor, processor));*/
        return stepBuilderFactory.get("step1")
                .<Person, Person>chunk(10)
                .reader(flatFileItemReader)
                .processor(processor)
                .writer(jdbcBatchItemWriter)
                .build();
    }
    // end::jobstep[]
}